/**
 * <p>Title: RocketMQService.java</p>
 * <p>Description: </p>
 * <p>Company: www.huoyunren.com</p>
 * @author duanxiangyun
 * @date 2016年11月24日
 * @version 1.0
 */
package com.smarttest.common.util;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Logger;

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.alibaba.rocketmq.tools.command.CommandUtil;

public class RocketMQService {
	private DefaultMQAdminExt adminExt;
	Logger log = Logger.getLogger(getClass());

	public RocketMQService(String adminExtGroup, String nameadd) {
		adminExt = new DefaultMQAdminExt(adminExtGroup);
		if (adminExt.getNamesrvAddr() == null
				|| adminExt.getNamesrvAddr().length() > 0) {
			adminExt.setNamesrvAddr(nameadd);
		}
		adminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
		try {
			adminExt.start();
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public RocketMQService(String nameadd) {
		adminExt = new DefaultMQAdminExt();
		if (adminExt.getNamesrvAddr() == null
				|| adminExt.getNamesrvAddr().length() > 0) {
			adminExt.setNamesrvAddr(nameadd);
		}
		adminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
		try {
			adminExt.start();
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}
	/**
	 * 通过topic、key从队列获取MQ消息
	 * 
	 * @param topic
	 * @param key
	 * @param count
	 *            获取条数（最大64条）
	 * @param starttime
	 *            开始时间
	 * @param endtime
	 *            结束时间
	 * @return 消息列表
	 */
	public List<MessageExt> getMessageExt(String topic, String key, int count,
			long starttime, long endtime) {
		List<MessageExt> l = new ArrayList<MessageExt>();
		try {
			QueryResult result = adminExt.queryMessage(topic, key, count,
					starttime, endtime);
			l = result.getMessageList();
			System.out.println("MQ nameserver地址==>"+adminExt.getNamesrvAddr()+",TOPIC==>"+topic+",KEY==>"+key+":");
			for (int i = 0; i < l.size(); i++) {
				MessageExt m = l.get(i);
				String msg = new String(m.getBody());
				System.out.println(msg);
			}

		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			if (e.getErrorMessage().contains(
					"query operation over, but no message")) {
				return new ArrayList<MessageExt>();
			}
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			close();
		}
		return l;
	}
	/**
	 * 通过topic、key从队列获取MQ消息
	 * 
	 * @param topic
	 * @param key
	 * @param count
	 *            获取条数（最大64条）
	 * @param starttime
	 *            开始时间
	 * @param endtime
	 *            结束时间
	 * @return 消息列表
	 */
	public List<String> getMQMsg(String topic, String key, int count,
			long starttime, long endtime) {
		List<String> msgs = new ArrayList<String>();
		try {
			QueryResult result = adminExt.queryMessage(topic, key, count,
					starttime, endtime);
			List<MessageExt> l = result.getMessageList();
			System.out.println("MQ nameserver地址==>"+adminExt.getNamesrvAddr()+",TOPIC==>"+topic+",KEY==>"+key+":");
			for (int i = 0; i < l.size(); i++) {
				MessageExt m = l.get(i);
				String msg = new String(m.getBody());
				msgs.add(msg);
//				System.out.println(msg);
				System.out.println("offset:"+m.getQueueOffset()+"storetime:"+m.getStoreTimestamp()+":"+msg);

			}

		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			if (e.getErrorMessage().contains(
					"query operation over, but no message")) {
				return new ArrayList<String>();
			}
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			close();
		}
		return msgs;
	}
	/**
	 * 根据topic、key获取MQ消息
	 * @param topic
	 * @param key
	 * @param count
	 * @return List<String>
	 */
	public List<String> getMQMsg(String topic, String key, int count) {
		List<String> msgs = new ArrayList<String>();
		try {
			QueryResult result = adminExt.queryMessage(topic, key, count, 0,
					Long.MAX_VALUE);
			List<MessageExt> l = result.getMessageList();
			System.out.println("MQ nameserver地址==>"+adminExt.getNamesrvAddr()+",TOPIC==>"+topic+",KEY==>"+key+":");
			for (int i = 0; i < l.size(); i++) {
				MessageExt m = l.get(i);
				String msg = new String(m.getBody());
				msgs.add(msg);
				System.out.println("offset:"+m.getQueueOffset()+"storetime:"+m.getStoreTimestamp()+":"+msg);
			}

		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			if (e.getErrorMessage().contains(
					"query operation over, but no message")) {
				return new ArrayList<String>();
			}
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			close();
		}
		return msgs;
	}

	/**
	 * 关闭
	 */
	public void close() {
		if (adminExt != null) {
			adminExt.shutdown();
		}
	}

	/**
	 * 获得队列最后消息入库时间
	 * @param topic
	 * @param key
	 * @return long
	 * @throws RemotingException 
	 * @throws MQBrokerException 
	 */
	public long getLastestTime(String topic, String key) {
		QueryResult result = null;
		try {
			result = adminExt.queryMessage(topic, key, 64, 0, Long.MAX_VALUE);
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			if (e.getMessage().contains("query operation over, but no message")) {
				return -1;
			} else {
				e.printStackTrace();
				close();
				log.error("MQ查询异常", e);
			}
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			close();
			log.error("MQ消息查询被中断", e);
		}
		List<MessageExt> list = result.getMessageList();
		long l = 0l;
		for (MessageExt msg : list) {
			Long l1 = msg.getStoreTimestamp();
			if (l < l1) {
				l = l1;
			}
		}
		return l;
	}

	public static void main(String[] args) {
//		 List<String> l =
//		 RocketMQQueryUtils.QueryMsgByKeySubCommand("172.16.1.44:9876",
//		 "topic_gps_trace", "107015090000099", 5, 0, Long.MAX_VALUE);	
		RocketMQQueryUtils.getLastestStorageTime("172.16.18.102:9876",
			     "topic_ctbox12_alarm", "3130F08060B1D30");
		 List<String> l =
			     RocketMQQueryUtils.QueryMsgByKeySubCommand("172.16.18.102:9876",
			     "topic_ctbox12_alarm", "3130F08060B1D30", 64);  
	}

	/**
	 * 删除消费组及关联关系
	 * @param clusterName cluster名称
	 * @param consumerGroup 消费组
	 * @throws InterruptedException
	 * @throws MQBrokerException
	 * @throws RemotingException
	 * @throws MQClientException
	 */
	public void deleteSubscriptionGroup(String clusterName, String consumerGroup)
			throws InterruptedException, MQBrokerException, RemotingException,
			MQClientException {
		Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(
				adminExt, clusterName);
		for (String master : masterSet) {
			adminExt.deleteSubscriptionGroup(master, consumerGroup);
			System.out
					.printf("delete subscription group [%s] from broker [%s] in cluster [%s] success.%n",
							consumerGroup, master, clusterName);
		}
		deleteTopic(adminExt, clusterName, MixAll.RETRY_GROUP_TOPIC_PREFIX
				+ consumerGroup);
		deleteTopic(adminExt, clusterName, MixAll.DLQ_GROUP_TOPIC_PREFIX
				+ consumerGroup);
		System.out.println("删除完毕：" + consumerGroup);
	}

	/**
	 * 删除topic
	 * @param adminExt DefaultMQAdminExt对象
	 * @param clusterName  cluster名称
	 * @param topic topic名称
	 * @throws InterruptedException
	 * @throws MQBrokerException
	 * @throws RemotingException
	 * @throws MQClientException
	 */
	public void deleteTopic(final DefaultMQAdminExt adminExt,//
			final String clusterName,//
			final String topic//
	) throws InterruptedException, MQBrokerException, RemotingException,
			MQClientException {

		Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(
				adminExt, clusterName);
		adminExt.deleteTopicInBroker(masterSet, topic);
		System.out.printf("delete topic [%s] from cluster [%s] success.%n",
				topic, clusterName);

		Set<String> nameServerSet = null;
		if (adminExt.getNamesrvAddr() != null) {
			String[] ns = adminExt.getNamesrvAddr().trim().split(";");
			nameServerSet = new HashSet(Arrays.asList(ns));
		}

		adminExt.deleteTopicInNameServer(nameServerSet, topic);
		System.out
				.printf("delete topic [%s] from NameServer success.%n", topic);
	}
}
